<?php

class MapRedActivity extends AsyncActivity {
	private $dynamicVariables;
	private $hdfsUtil;

    public function __construct($activity) {
   		parent::__construct($activity);  
		$this->hdfsUtil = new HdfsUtil("{$this->hdfsSystem['client_home']}/bin/hadoop");
		$this->dynamicVariables = array(); 
    }
    
    protected function checkDepends() {
    	// Check the depend resource partition is ready
		if(!parent::checkDepends()) {
			return false;
		}
		foreach($this->inputs as $input) {
			Yii::log("check input resource[{$input['resource']['physical']['physical_name']}]", CLogger::LEVEL_INFO, WORKER_LOG);
			
			foreach ($input['partitions'] as $partition) {
				if($input['input_type'] == ResourceType::Log) {
					$status = $this->hdfsUtil->exists("{$input['resource']['physical']['storage']}/$partition");
				}
				else {
					$status = $this->resourceService->checkResourcePartition($input['resource']['physical']['physical_id'], $partition);
				}
				if(!$status) {
					Yii::log("physical resource[{$input['resource']['physical']['physical_name']}] partiion[$partition] is not ready", CLogger::LEVEL_ERROR, WORKER_LOG);
					
					$this->setRuntimeInfo(array(
						'state'	=>	TaskState::Waiting,
						'message'	=>	"[MapRed/checkDepend]physical resource[{$input['resource']['physical']['physical_name']}] partiion[$partition] is not ready",
					));
					return false;
				}
			}
		}
		return true;
    }
	
	protected function preparePrograms() {
		$programs = CJSON::decode($this->activity['programs']);
    	    	
    	if(count($programs) == 2) {
	    	$map = $programs[0];
	    	$reduce = $programs[1];
	    	$jar = " jar {$this->hdfsSystem['client_home']}/contrib/streaming/hadoop-streaming.jar";
	    	$mapper  = ' -mapper '.basename($map);
        	$reducer = ' -reducer '.basename($reduce);
        	$fileStr = " -file $map -file $reduce";
    	}
    	else if(count($programs) == 1){
    		$jar = " jar $programs[0]";
    	}
    	else {
    		Yii::log('the programs number is wrong', CLogger::LEVEL_ERROR, WORKER_LOG);
    		return false;
    	}
    	
    	$params = CJSON::decode($this->activity['params']);
    	if(isset($params) && is_array($params)) {
    		$ret = $this->parseParams($params);
    		if($ret === false) {
    			return false;
    		}
    		$define .= $ret['define'];
    		$fileStr .= $ret['file'];
    		$inputFormat = $ret['inputformat'];
    		$outputFormat = $ret['outputformat'];
    		$partitioner = $ret['partitioner'];
    	}
    	
    	$relyFileStr = $this->scanRelyFiles();
    	
        foreach($this->inputs as $input) {        	
        	foreach ($input['partitions'] as $partition) {
        		$physicalType = $this->getPhysicalType($input['input_type']);
        		if($physicalType != PhysicalType::Hive) {
        			$inputStr .= " -input {$input['resource']['physical']['storage']}/$partition";
        		}
				else {
					$inputStr .= " -input {$input['resource']['physical']['storage']}";
				}
        		if(!empty($input['log_pattern'])) {
        			$inputStr .= '/' . trim($input['log_pattern']);
        		}
        	}
        }
        
        $output = $this->outputs[0];
        $partition = $output['partitions'][0];
    	$outputStr = " -output {$output['resource']['physical']['storage']}/$partition";
    	$this->hdfsUtil->rm("{$output['resource']['physical']['storage']}/$partition", true);
        $programs[0] = "{$this->hdfsSystem['client_home']}/bin/hadoop" . $jar . $define . $partitioner . $mapper . $reducer . 
        	$inputFormat . $outputFormat . $inputStr . $outputStr . $fileStr . $relyFileStr;
    	return $programs;
	}
	
	protected function save() {		
		if(!parent::save()) {
			return false;
		}
		
		Yii::log("save output resource:[{$output['resource']['res_name']}]", CLogger::LEVEL_INFO, WORKER_LOG);
		$output = $this->outputs[0];				
		$partition = $output['partitions'][0];
		$outputPath = "{$output['resource']['physical']['storage']}/$partition";
		
		if($output['multiple']) {
			$files = $this->hdfsUtil->readDir($outputPath);
			foreach ($files as $file) {
				list($type) = explode("-", basename($file->path));
				$types[$type][] = $file->path;
			}

			foreach ($types as $type => $paths) {
				if(!$this->hdfsUtil->mkdir("$outputPath/$type")) {
					Yii::log("make hdfs dir[$outputPath/$type] failed", CLogger::LEVEL_ERROR, WORKER_LOG);
					$this->setRuntimeInfo(array(
						'state'	=>	TaskState::Failed,
						'message'	=>	"[MapRed/save]make hdfs dir[{$outputPath}/{$type}] failed",
					));
					return false;
				}
				
				foreach($paths as $path) {
					if($this->hdfsUtil->rename($path, "$outputPath/$type")) {
						Yii::log("rename hdfs file from[$path] to[{$outputPath}/{$type}] failed", CLogger::LEVEL_ERROR, WORKER_LOG);
						$this->setRuntimeInfo(array(
							'state'	=>	TaskState::Failed,
							'message'	=>	"[MapRed/save]rename hdfs file from[$path] to[{$outputPath}/{$type}] failed",
						));
						return false;
					}
				}
				if(!$this->addOutputPartition($output, "$partition/$type")) {
					return false;
				}
			}
			return true;
		}
		
		return $this->addOutputPartition($output, $partition);
    }
    
	protected function getStatus() {
		// check hadoop _success file exist or not
		$output = $this->outputs[0];
		$partition = $output['partitions'][0];
		$successFile = $output['resource']['physical']['storage'] . '/' . $partition . '/_SUCCESS';
		$status = PathUtil::hdfsFileExists($this->hdfsSystem['client_home'], $successFile);
		if($status) {
			Yii::log("hdfs file [$successFile] exists", CLogger::LEVEL_INFO, WORKER_LOG);
		}
		else {
			Yii::log("hdfs file [$successFile] doesn't exists", CLogger::LEVEL_ERROR, WORKER_LOG);
		}
		return $status;
	}
	
	private function addOutputPartition($output, $partition) {
		// Save the resource partition.
		$partitionRecord = array(
			'physical_id'	=>	$output['resource']['physical']['physical_id'],
			'partition'		=>	$partition,
			'size'			=>	$this->getResultSize(),
			'count'			=>	$this->getResultCount(),
			'schema'		=>	'',
			'create_time'	=>	date('Y-m-d H:i:s'),
		); 
		
		$partitionId = $this->resourceService->addResourcePartition($partitionRecord);
		if($partitionId === false) {
			Yii::log("add resource[{$output['resource']['res_name']}] partition failed", CLogger::LEVEL_ERROR, WORKER_LOG);
			$this->setRuntimeInfo(array(
				'state'	=>	TaskState::Failed,
				'message'	=>	"[MapRed/save]add resource[{$output['resource']['res_name']}] partition failed",
			));
			return false;
		}
		Yii::log("add resource[{$output['resource']['res_name']}] partition[] successfully", CLogger::LEVEL_INFO, WORKER_LOG);
		return true;
	}
    
    private function scanRelyFiles() {
    	$filePath = $this->env['runtime'] . '/file';
    	$relyFiles = scandir($filePath);
    	if(is_array($relyFiles)) {
    		foreach($relyFiles as $relyFile) {
    			if($relyFile == '.' || $relyFile == '..') {
    				continue;
    			}
    			$relyFileStr .= " -file $filePath/$relyFile";
    		}
    	}
    	return $relyFileStr;
    }
    
    private function replaceVariable($str, $variable) {
    	$iniPath = $this->env['runtime'] . '/variable/variable.ini';
    	if(!file_exists($iniPath)) {
    		return false;
    	} 
    	
    	if(count($this->dynamicVariables) == 0) {
    		$this->dynamicVariables = parse_ini_file($iniPath);
    	}
    	
    	if(!isset($this->dynamicVariables[$variable])) {
    		return false;
    	}
    	
    	return preg_replace('/\${\w+}/', $this->dynamicVariables[$variable], $str);    	    	
    }
    
    private function parseParams($params) {
    	$ret = array();
    	foreach ($params as $param) {
    		if(StringUtil::startsWith($param, '-D')) {
    			if(preg_match('/\${?(\w+)}/', $param, $matches)) {
    				$param = $this->replaceVariable($param, $matches[1]);
    			}
    			if($param === false) {
    				Yii::log("replace variable[{$matches[1]}] failed", CLogger::LEVEL_ERROR, WORKER_LOG);
    				$this->setRuntimeInfo(array(
						'state'	=>	TaskState::Failed,
						'message'	=>	"[MapRed/prepare]replace variable[{$matches[1]}] failed",
					));
    				return false;
    			}
    			$ret['define'] .= " $param";
    		}
    		if(StringUtil::startsWith($param, '-jobconf')) {
    			if(preg_match('/\${?(\w+)}/', $param, $matches)) {
    				$param = $this->replaceVariable($param, $matches[1]);
    			}
    			if($param === false) {
    				return false;
    			}
    			$ret['define'] .= " $param";
    		}
    		elseif(StringUtil::startsWith($param, '-file')) {
    			$ret['file'] .= " $param";
    		}
    		elseif(StringUtil::startsWith($param, '-cacheFile')) {
    			$ret['file'] .= " $param";
    		}
    		elseif(StringUtil::startsWith($param, '-files')) {
    			$ret['file'] .= " $param";
    		}
    		elseif(StringUtil::startsWith($param, '-cacheArchive')) {
    			$ret['file'] .= " $param";
    		}
    		elseif(StringUtil::startsWith($param, '-archives')) {
    			$ret['file'] .= " $param";
    		}
    		elseif(StringUtil::startsWith($param, '-inputformat')) {
    			$ret['inputformat'] .= " $param";
    		}
    		elseif(StringUtil::startsWith($param, '-outputformat')) {
    			$ret['outputformat'] .= " $param";
    		}
    		elseif(StringUtil::startsWith($param, '-partitioner')) {
    			$ret['partitioner'] .= " $param";
    		}
    	}
    	return $ret;
    }
}
